package com.irt.rootcloud.realtime.gketl.bolt;

import java.util.Map;
import java.util.Properties;

import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

import com.alibaba.fastjson.JSONObject;
import com.irt.rootcloud.realtime.gketl.util.Utils;

/**
 * 
 * @author Ken.zhang
 * @date 2017-1-18 下午2:30:02 发送工况数据到es
 */
public class EsSendBolt extends BaseBasicBolt {

	/**
	 * 
	 */
	private static final long serialVersionUID = -8978478582442232414L;
	private Properties properties;
	private static Client client;
	private String tennat;
	private String deviceType;
	private BulkRequestBuilder bulkRequest;
	private long count = 0;

	public EsSendBolt(Properties properties) {
		this.properties = properties;
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		// TODO Auto-generated method stub
		super.prepare(stormConf, context);
		initEsClient();
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		super.cleanup();
	}

	private void initEsClient() {
		Settings settings = ImmutableSettings.settingsBuilder()
				// 客户端嗅探整个集群的状态，把集群中其它机器的ip地址自动添加到客户端中，并且自动发现新加入集群的机器
				.put("client.transport.sniff", true).put("client", true)
				.put("data", false)// 仅作为客户端连接
				.put("cluster.name", properties.getProperty("es.cluster.name"))// 集群名称"elasticsearch"
				.build();
		client = new TransportClient(settings)
				.addTransportAddress(new InetSocketTransportAddress(properties
						.getProperty("es.ip"), Integer.valueOf(properties
						.getProperty("es.ip.port"))));
		bulkRequest = client.prepareBulk();
	}

	/*private void sendEsMessage(String tennat, String deviceType,
			JSONObject jsonobject) {
		IndexResponse response = client.prepareIndex(tennat, deviceType)
				.setSource(jsonobject).execute().actionGet();
	}*/

	public void execute(Tuple input, BasicOutputCollector collector) {
		try {
			JSONObject dtjson = JSONObject.parseObject(input
					.getValueByField(Utils.FIELDCHANNEL_BOLT).toString());
			tennat = (String) input.getValueByField(Utils.CHANNEL_TENANT);
			deviceType = (String) input
					.getValueByField(Utils.CHANNEL_DEVICETYPE);
//			sendEsMessage(tennat, deviceType, dtjson);
			//1000条提交一次
			bulkRequest.add(client.prepareIndex(tennat, deviceType).setSource(dtjson));
			if (count% 1000==0) {
                bulkRequest.execute().actionGet();
                System.out.println("提交了：" + count);
                count = 0;
            }
			count++;
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
	}

}
